/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.googlecode.j4sc.submission.manager;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.googlecode.j4sc.conversion.Conversion;
import com.googlecode.j4sc.service.IService;
import com.googlecode.j4sc.service.ServiceException;
import com.googlecode.j4sc.submission.Submission;
import com.googlecode.j4sc.submission.SubmissionReference;

/**
 * The SubmissionManager is used to simplify submitting of {@link Submission}'s
 * and in conjunction with a {@link RetrievalStrategy}, will take care of
 * retrieval of conversions. Upon a {@link Conversion} which is not pending
 * being retrieved, it will notify all {@link ConversionListener}'s which have
 * been registered using the {@link #addListener(ConversionListener)} method.
 * <p>
 * If the {@link RetrievalStrategy} in use throws an
 * {@link ExpiredSubmissionReferenceException} the {@link SubmissionManager}
 * will call the {@link ConversionListener#expired(SubmissionReference)} method
 * on any registered {@link ConversionListener}'s.
 * <p>
 * The example code below shows how to use the SubmissionManager.
 * <p>
 * 
 * <pre>
 * public class SubmissionManagerExample implements ConversionListener {
 *     public static void main(String[] args) {
 * 	int maxRetrievalAttempts = 4;
 * 	SubmissionManager manager = new SubmissionManager(Service.LIVE,
 * 			new BasicRetrievalStrategy(maxRetrievalAttempts));
 * 	try {
 * 		manager.addListener(new SubmissionManagerExample());
 * 		manager.startRetrieving();
 * 		Submission submission = new Submission(&quot;account id&quot;, &quot;app name&quot;,
 * 				&quot;reference&quot;, new Audio(new File(&quot;/tmp/audio.wav&quot;), 
 * 				new UserCredentials(&quot;user&quot;, &quot;pass&quot;));
 * 		manager.submit(submission);
 * 		while(true){
 * 			// wait while conversion is retrieved
 * 		}
 * 	} catch (Exception e) {
 * 		e.printStackTrace();
 * 	}
 * }
 * 
 *     public void exception(SubmissionReference reference, Exception exception) {
 *         exception.printStackTrace();
 *     }
 * 
 *     public void expired(SubmissionReference reference) {
 *         System.out.println(&quot;Submission with reference [&quot; + reference.getReference() + &quot;] has expired&quot;);
 *     }
 * 
 *     public void received(Conversion conversion) {
 *         System.out.println(&quot;Status: &quot; + conversion.getStatus());
 *         System.out.println(&quot;Text: &quot; + conversion.getText());
 *     }
 * }
 * </pre>
 * 
 * @author Raj Patel, Dariusz Lorenc
 */
public class SubmissionManager {

    private static final int JOB_EXECUTION_PERIOD_IN_SECONDS = 5;

    private final IService service;
    private final Set<SubmissionReference> submissionReferences;
    private final List<ConversionListener> listeners;
    private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();

    private final RetrievalStrategy retrievalStrategy;

    /**
     * Construct a new SubmissionManager
     * 
     * @param service
     *            the underlying {@link IService} to use for submitting and
     *            retrieving
     * @param retrievalStrategy
     *            the {@link RetrievalStrategy} to use to when retrieving
     */
    public SubmissionManager(IService service, RetrievalStrategy retrievalStrategy) {
        if (service == null) {
            throw new IllegalArgumentException("Service must not be null");
        }
        if (retrievalStrategy == null) {
            throw new IllegalArgumentException("RetrievalStrategy must not be null");
        }
        this.retrievalStrategy = retrievalStrategy;
        this.service = service;
        this.submissionReferences = new LinkedHashSet<SubmissionReference>();
        this.listeners = new ArrayList<ConversionListener>();
    }

    /**
     * Submit a {@link Submission} to the service for conversion
     * 
     * @param submission
     *            the submission
     * @throws ServiceException
     *             thrown if the service responds with an error
     * @throws IOException
     *             thrown on I/O errors
     */
    public void submit(Submission submission) throws ServiceException, IOException {
        submissionReferences.add(service.submit(submission));
    }

    /**
     * Gets the set of {@link SubmissionReference}'s being managed
     * 
     * @return set of {@link SubmissionReference}'s
     */
    public Set<SubmissionReference> getSubmissionReferences() {
        return Collections.unmodifiableSet(submissionReferences);
    }

    /**
     * Remove a {@link SubmissionReference} from being retrieved. This method is
     * executed on best endeavours as the {@link SubmissionReference} may be in
     * the process of being retrieved at the time.
     * 
     * @param submissionReference
     */
    public void removeSubmissionReference(SubmissionReference submissionReference) {
        submissionReferences.remove(submissionReference);
    }

    /**
     * Add a {@link ConversionListener} which will be notified of events.
     * Multiple {@link ConversionListener}'s can be added, of which all will be
     * notified upon an event
     * 
     * @param listener
     *            the {@link ConversionListener}
     */
    public void addListener(ConversionListener listener) {
        listeners.add(listener);
    }

    /**
     * Start retrieving
     */
    public void startRetrieving() {
        startRetrieving(JOB_EXECUTION_PERIOD_IN_SECONDS, TimeUnit.SECONDS);
    }

    void startRetrieving(int interval, TimeUnit unit) {
        executorService.scheduleWithFixedDelay(new RetrievalJob(), interval, interval, unit);
    }

    boolean doesNotContain(SubmissionReference reference) {
        return !submissionReferences.contains(reference);
    }

    private class RetrievalJob implements Runnable {

        public void run() {
            Set<SubmissionReference> references = new LinkedHashSet<SubmissionReference>(submissionReferences);
            List<Thread> createdThreads = new ArrayList<Thread>();
            for (SubmissionReference reference : references) {
                try {
                    if (retrievalStrategy.isDueForRetrieval(reference)) {
                        Thread thread = new Thread(new SubmissionRetriever(reference));
                        thread.start();
                        createdThreads.add(thread);
                    }
                } catch (ExpiredSubmissionReferenceException e) {
                    removeSubmissionReference(reference);
                    for (ConversionListener listener : listeners) {
                        listener.expired(reference);
                    }
                }
            }

            waitForThreadsToComplete(createdThreads);
        }

        private void waitForThreadsToComplete(List<Thread> threads) {
            for (Thread thread : threads) {
                try {
                    thread.join();
                } catch (InterruptedException e) {
                }
            }
        }
    }

    private class SubmissionRetriever implements Runnable {

        private final SubmissionReference reference;

        public SubmissionRetriever(SubmissionReference reference) {
            this.reference = reference;
        }

        public void run() {
            try {
                Conversion conversion = service.retrieve(reference);
                if (!conversion.isPending()) {
                    removeSubmissionReference(reference);
                    for (ConversionListener listener : listeners) {
                        listener.received(conversion);
                    }
                }
            } catch (Exception e) {
                for (ConversionListener listener : listeners) {
                    listener.exception(reference, e);
                }
            }
        }
    }
}
